package extractor;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Comparator;
import java.util.Iterator;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.FileOutputFormat;
import org.apache.hadoop.mapred.JobClient;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.MapReduceBase;
import org.apache.hadoop.mapred.Mapper;
import org.apache.hadoop.mapred.OutputCollector;
import org.apache.hadoop.mapred.Reducer;
import org.apache.hadoop.mapred.Reporter;
import org.apache.hadoop.mapred.TextInputFormat;
import org.apache.hadoop.mapred.lib.MultipleInputs;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

import utils.Utils;
import fe.FeatureConfig;
import fe.FeatureManager;

public class FeatureExporter extends Configured implements Tool{
	public static String getCPFeaturePath(String date) {
		return Utils.rootDir + date + "/CPFeature/";
	}
	
	enum Counter{INVALID_MAP_INPUT, EXTRACT_ERROR};
	
	public static class Map extends MapReduceBase implements Mapper<LongWritable, Text, Text, Text> {
		private Text k = new Text();
		public void map(LongWritable key, Text value, 
				OutputCollector<Text, Text> output, Reporter reporter) throws IOException {
			String[] values = value.toString().split(Utils.separator);
			if (values.length != 130) {
				reporter.incrCounter(Counter.INVALID_MAP_INPUT, 1);
				return;
			}
			BaseRecord record = new BaseRecord(values);
			k.set(record.getUUID());
			output.collect(k, value);
		}
	}
	
	public static class ItemComparator implements Comparator<BaseRecord> {

		@Override
		public int compare(BaseRecord arg0, BaseRecord arg1) {
			// TODO Auto-generated method stub
			
			return arg0.getItemPosition() - arg1.getItemPosition();
		}
		
	}
	public static class Reduce extends MapReduceBase implements Reducer<Text, Text, Text, Text> {
		private Text k = new Text();
		private Text v = new Text();
		@Override
		public void configure(JobConf job) {
			String featureConfig = job.get(FeatureConfig.FEATURE_CONFIG_NAME);
			try {
				featureManager.initManager(featureConfig, FeatureConfig.FEATURE_CONFIG_SEPARATOR);
			} catch (Exception e) {
				System.out.println(e.getMessage());
				featureManager = null;
			}
		}
		private FeatureManager featureManager = new FeatureManager();
		@Override
		public void reduce(Text key, Iterator<Text> iter,
				OutputCollector<Text, Text> output, Reporter reporter)
				throws IOException {
			// TODO Auto-generated method stub
			ArrayList<BaseRecord> records = new ArrayList<BaseRecord>();
			while(iter.hasNext()) {
				BaseRecord record = new BaseRecord(iter.next().toString().split(Utils.separator));
				records.add(record);
			}
			java.util.Collections.sort(records, new ItemComparator());
			featureManager.beginRequest();
			for (int i = 0; i < records.size(); ++i) {
				BaseRecord record = records.get(i);
				k.set(record.getBuyer32Id());
				try {
					v.set(featureManager.getFeaturesStr(records, i));
				} catch(Exception e) {
					reporter.incrCounter(Counter.EXTRACT_ERROR, 1);
					System.out.println(e.getMessage());
					e.printStackTrace();
					continue;
				}
				output.collect(k, v);
			}
			featureManager.endRequest();
		}
		
	}
	@Override
	public int run(String[] args) throws Exception {

		// input
		String baseItemFile = AttachItemCoec.getItemCoecPath(args[0]);		

		// Output
		String output = getCPFeaturePath(args[0]);
		System.out.println("start job ExportClickPredictionFeature...");
		System.out.println("baseItemFile: " + baseItemFile);
		System.out.println("output: " + output);
		if (Utils.exitSystem(args))
			return -1;
		
		JobConf conf = new JobConf(getConf(), FeatureExporter.class);
		conf.setJobName("ExportClickPredictionFeature");
		conf.setOutputKeyClass(Text.class);
		conf.setOutputValueClass(Text.class);
		
		MultipleInputs.addInputPath(conf, 
				new Path(baseItemFile),
				TextInputFormat.class, 
				Map.class);

		FeatureConfig featureConfig = new FeatureConfig(FeatureConfig.DEFAULT_CONFIG_PATH);
	
		FileOutputFormat.setOutputPath(conf, new Path(output));
		conf.set(FeatureConfig.FEATURE_CONFIG_NAME, featureConfig.getFeatureConfig());

		conf.setReducerClass(Reduce.class);

		JobClient.runJob(conf);
		return 0;
	}
	
		
	public static void main(String[] args) throws Exception {
		Utils.checkArgs(args);
		
		int res = ToolRunner.run(new Configuration(), new FeatureExporter(),
				args);
		System.exit(res);
	}

}
